package com.mutouren.modules.datapush.archives;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.apache.commons.lang3.time.DateUtils;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Service;

import com.mutouren.common.log.LogAlias;
import com.mutouren.common.log.LogManager;
import com.mutouren.common.log.Logger;
import com.mutouren.common.thread.ThreadPoolManager;
import com.mutouren.modules.datapush.base.DataPushLogAlias;
import com.mutouren.modules.datapush.base.PushMessage;
import com.mutouren.modules.datapush.dao.PushMessageDao;
import com.mutouren.modules.datapush.dao.impl.DaoFactoryImpl;
import com.mutouren.modules.datapush.remote.RemotePushServer;

@Service
public class Sender implements ApplicationContextAware {
	
	private final static int pullSize = 200;
	private final static int searchLastHour = 72;
	
	//@Resource
	private PushMessageDao pushMessageDao = DaoFactoryImpl.getInstance().getPushMessageDao();
	
	@Resource(name="remotePushServer")
	private RemotePushServer pushServer;
	
	private ExecutorService leader;
	
	private Object waitTaskEvent = new Object();
	
	private boolean isDataBaseException = false;
	private long searchIndex = 0;
	
	private volatile boolean isOpened = false;
	
	protected final static Logger monitorLogger = LogManager.getLogger(DataPushLogAlias.DataPushMonitor.name());	
	protected final static Logger errorLogger = LogManager.getLogger(LogAlias.SystemError.name());
	protected final static Logger runLogger = LogManager.getLogger(LogAlias.SystemRun.name());	
	
	public Sender() {
		leader = ThreadPoolManager.createSingleThreadExecutor();
	}
	
	public synchronized void open() {
		if (!this.isOpened) {
			pushServer.open("");
			this.isOpened = true;
			leader.execute(()-> this.search());
		}
	}
	
	public synchronized void close() {
		pushServer.close();
		
		this.isOpened = false;
		leader.shutdownNow();
		
		try {
			leader.awaitTermination(5, TimeUnit.SECONDS);		
		} catch (Exception e) {
			errorLogger.error("Sender server awaitTermination error", e);
		}
	}	
	
	private long lastMoniterTime = 0;
	private volatile long totalNum = 0;
	private void debugMoniter(int size, boolean isPermitPull) {
		long curTime = new Date().getTime();		
		if ((curTime - lastMoniterTime) > 5000) {		
			String log = String.format("Sender server pull size:%d, isPermitPull:%s, isDataBaseException:%s,totalNum:%d", 
					size, isPermitPull, isDataBaseException, totalNum);
			monitorLogger.debug(log);
			lastMoniterTime = curTime;
		}
	}	

	private void search() {
		do {
			int curPullCount = 0;
			boolean isPermitPull = this.isPermitPull();
			if (isPermitPull) {
				List<PushMessage> message = pollMessageFromDB();
				curPullCount = message.size();
				totalNum +=curPullCount;
				processMessage(message);
			}
			
			debugMoniter(curPullCount, isPermitPull);			
			if ((curPullCount == pullSize)&&(this.isOpened)&&(!this.isDataBaseException)) continue;
							
			synchronized(this.waitTaskEvent) {
				try {
					waitTaskEvent.wait(1000);
				} catch (InterruptedException e) {
					//?errorLogger.warn("Sender search Interrupted", e);
				}
			}
			
		} while(ThreadPoolManager.isRun()&&this.isOpened);		
	}	
	
	private List<PushMessage> pollMessageFromDB() {
		List<PushMessage> result;
		try {
			if (this.searchIndex <= 0) {
				Date indexTime = DateUtils.addHours(new Date(), -searchLastHour);
				result = pushMessageDao.selectByCreateTime(indexTime, pullSize);
			} else {
				result = pushMessageDao.selectById(this.searchIndex, pullSize);
			}
			
			if (result.size() > 0) {
				this.searchIndex = result.get(result.size() - 1).getId();
			}
			
			for(PushMessage message : result){
				message.thinData();
			}
			isDataBaseException = false;	
		} catch(Throwable t) {
			isDataBaseException = true;	
			result = new ArrayList<PushMessage>();
			errorLogger.error("Sender server pollMessageFromDB error", t);
		}

		return result;
	}
	
	private void processMessage(List<PushMessage> listMessage) {
		for(PushMessage message : listMessage) {
			send(message);
		}
	}
	
	private boolean isPermitPull() {
		return pushServer.isPermitReceive(pullSize);
	}
		
	private void send(PushMessage message) {
		pushServer.receive(message);		
	}

	public void pullNotify() {
		synchronized(this.waitTaskEvent) {
			this.waitTaskEvent.notify();
		}	
	}

	@Override
	public void setApplicationContext(ApplicationContext context) throws BeansException {
//		System.out.println("bean count : " + context.getBeanDefinitionCount());
//		for(String beanName : context.getBeanDefinitionNames()) {
//			System.out.println(beanName);
//		}				
	}
}
